package com.hrg.seer.sdk.rbk;

import com.hrg.rds.biz.config.RdsConfiguration;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.CompositeByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
class RbkPortClient {
    @Getter
    private final String host;
    @Getter
    private final Integer port;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();
    private volatile boolean resetting = true;
    private Integer flowNoCounter = 0;
    private final Map<Integer, String> responseMap = new ConcurrentHashMap<>();
    private volatile NettyClient nettyClient = null;
    // 用于异步创建 NettyClient 的线程池，用于连接失败时重试创建连接
    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();

    RbkPortClient(String host, Integer port) {
        this.port = port;
        this.host = host;
        getNettyClient();
    }

    RbkResult request(Integer apiNo, String reqStr, Long timeout) {
        RbkResult rbkResult = doRequest(apiNo, reqStr, timeout);
        if (rbkResult.getKind() != RbkResultKind.Ok) {
            log.info("请求失败，销毁 RBK 端口客户端，错误类型={}, apiNo={}", rbkResult.getKind(), apiNo);
            reset();
        }
        return rbkResult;
    }

    private RbkResult doRequest(int apiNo, String reqStr, Long timeout) {
        // 等待响应未回复，默认10s超时
        timeout = timeout == null ? 10 * 1000 : timeout;
        lock.lock();
        // - 每个端口 API 请求的时间间隔最快建议为 100～200ms，请勿过快。
        // - 机器人将对每个建立的 TCP 连接使用 TCP KeepAlive 机制进行保活, 对已死的连接会及时清除。
        // - 机器人若收到任何不符合本协议报文头部的错误数据包, 将会主动关闭该 TCP 连接（不会回复任何信息）。
        // - 其他情况下, 机器人不会主动断开连接, API 使用者应自己控制连接何时中断。
        // - 对于每个连接, 机器人都是以一问一答的形式进行处理的, 即未作出响应前是不会处理下一个请求的, 因此建议在使用时，先等待收到上一次请求的响应后再发送下一个请求。
        try {
            NettyClient c;
            RbkResult.RbkResultBuilder rbkResultBuilder = RbkResult.builder()
                    .ip(host).apiNo(apiNo)
                    .reqStr(reqStr).resStr("");
            try {
                c = getNettyClient();
            } catch (Exception e) {
                log.error("connect error ip {} port {} {}", host, port, e.getMessage());
                return e instanceof InterruptedException ? rbkResultBuilder.kind(RbkResultKind.Interrupted).build()
                        : rbkResultBuilder.kind(RbkResultKind.ConnectFail).errMsg(e.getMessage()).build();
            }
            Integer flowNo = nextFlowNo(); // 流水号
            CompositeByteBuf reqBuf = RbkEncoder.buildReqBytes(apiNo, reqStr, flowNo);
            log.debug("发送消息，ip = {}, port = {}, flowNo = {}, reqStr = {}", host, port, flowNo, reqStr);
            try {
                c.write(reqBuf);
            } catch (InterruptedException e) {
                log.error("doRequest write error ip {} port {} {}", host, port, e.getMessage());
                return rbkResultBuilder.kind(RbkResultKind.WriteError).errMsg(e.getMessage()).build();
            }

            while (!resetting) {
                try {
                    if (!condition.await(timeout, TimeUnit.MILLISECONDS)) {
                        //等待超时
                        return rbkResultBuilder.kind(RbkResultKind.Timeout).errMsg("Timeout").build();
                    }
                } catch (InterruptedException e) {
                    // 等待结果被中断
                    log.error("Waiting for results to be interrupted ip {} port {} {}", host, port, e.getMessage());
                    return rbkResultBuilder.kind(RbkResultKind.Interrupted).errMsg("Timeout").build();
                }
                if (responseMap.containsKey(flowNo)) {
                    String resStr = responseMap.remove(flowNo);
                    return rbkResultBuilder.kind(RbkResultKind.Ok).resStr(resStr).build();
                }
            }
            //已重置
            return rbkResultBuilder.kind(RbkResultKind.Disposed).build();
        } finally {
            lock.unlock();
        }

    }

    public NettyClient getNettyClient() {
        // 启用后才初始化以及重连
        NettyClient c = nettyClient;
        if (c == null) {
            synchronized (this) {
                if (c == null) {
                    try {
                        initNettyClient();
                    } catch (Exception e) {
                        log.error("getNettyClient error", e);
                    }
                }
            }
        }
        return c;
    }

    public void initNettyClient() {
        if (!RdsConfiguration.AgvIpMap.get(host).isIfEnabled()) return;
        // 创建客户端
        RbkInboundHandler handler = new RbkInboundHandler(this);
        NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000).option(ChannelOption.SO_KEEPALIVE, true);
        b.group(eventExecutors)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(host, port))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast(new RbkDecoder());
                        pipeline.addLast(handler);
                    }
                });
        try {
            ChannelFuture cf = b.connect().sync();
            this.nettyClient = NettyClient.builder().eventLoopGroup(eventExecutors).channelFuture(cf).build();
            this.resetting = false;
        } catch (Exception e) {
            log.error("initNettyClient error，{}", e.getMessage());
            eventExecutors.shutdownGracefully();
            //重连
            executorService.schedule(this::initNettyClient, 5L, TimeUnit.SECONDS);
        }
    }

    void onMessage(ChannelHandlerContext ctx, RbkFrame msg) {
        responseMap.put(msg.getFlowNo(), msg.getBodyStr());
        lock.lock();
        try {
            condition.signalAll();
        } finally {
            lock.unlock();
        }
    }

    // 只接收推送的消息
    void onPushedMessage(String bodyStr) {
        try {
            //log.debug("onPushedMessage, {}", bodyStr);
            RdsConfiguration.AgvIpMap.get(this.host).onPushedMessage(bodyStr);
        } catch (Exception e) {
            log.error("onPushedMessage error, {}", e.getMessage());
        }
    }

    void onError(Throwable e) {
        log.error("netty error, {}", e.getMessage());
        reset();
    }

    private Integer nextFlowNo() {
        Integer no = (flowNoCounter + 1) % 512;
        flowNoCounter = no;
        return no;
    }

    private synchronized void reset() {
        responseMap.clear();
        if (resetting || nettyClient == null) {
            return;
        }
        // logger.debug("dispose rbk client start")
        resetting = true;
        NettyClient c = nettyClient;
        try {
            c.close();
        } catch (Exception e) {
            // ignore
        }
        nettyClient = null;
    }

    void dispose() {
        reset();
    }
}
